﻿using RdKafka;
using System;
using System.Collections.Generic;
using System.Text;
using System.Linq;
using System.Threading;
using System.Net.Sockets;
using BCGPSCommon;
using System.Net;
using AppSocketModel;

namespace BCGPS.DataRealy.Service
{
    public class ReadKafakaConsumer
    {
        string brokerList = string.Empty;
        List<string> topicList;
        bool enableAutoCommit = false;
        Topic topic;
        EventConsumer eventConsumer = null;
        private Dictionary<int, UdpDataSendClient> udpDataSendClientDict = null;


        private void WriteLine(string msg)
        {
            Console.WriteLine(msg);
        }
        public ReadKafakaConsumer(Dictionary<int, UdpDataSendClient> udpDataSendClientDict)
        {
            string topicName = "BCGPS-Realy";
            brokerList = ConfigurationManager.AppSettings["Kafka:BrokerList"];
            topicList = new List<string>() { topicName };
            enableAutoCommit = bool.Parse(ConfigurationManager.AppSettings["Kafka:EnableAutoCommit"]);

            Config config = new Config()
            {
                GroupId = topicName,
                EnableAutoCommit = enableAutoCommit,
                StatisticsInterval = TimeSpan.FromSeconds(60),
            };

            eventConsumer = new EventConsumer(config, brokerList);

            this.udpDataSendClientDict = udpDataSendClientDict;
        }

        public void Start()
        {
            eventConsumer.OnMessage += EventConsumer_OnMessage;

            eventConsumer.OnEndReached += EventConsumer_OnEndReached;
            eventConsumer.OnConsumerError += EventConsumer_OnConsumerError;
            eventConsumer.OnError += EventConsumer_OnError;
            eventConsumer.OnPartitionsAssigned += EventConsumer_OnPartitionsAssigned;
            eventConsumer.OnPartitionsRevoked += EventConsumer_OnPartitionsRevoked;
            //eventConsumer.OnStatistics += EventConsumer_OnStatistics;

            if (enableAutoCommit)
            {
                eventConsumer.OnOffsetCommit += EventConsumer_OnOffsetCommit;
            }
            eventConsumer.Subscribe(topicList);
            eventConsumer.Start();
        }

        private void stop()
        {
            eventConsumer.Stop();
            Thread.Sleep(1000);
        }

        #region   消息处理
        private void EventConsumer_OnOffsetCommit(object sender, Consumer.OffsetCommitArgs commit)
        {
            string str = string.Empty;
            if (commit.Error != ErrorCode.NO_ERROR)
            {
                str = $"  Failed to commit offsets:{commit.Error}";
                WriteLine(str);

            }
            str = $"  Sussessfully committed offsets:[{string.Join(", ", commit.Offsets)}]";
            WriteLine(str);
        }

        private void EventConsumer_OnStatistics(object sender, string json)
        {
            string str = string.Empty;
            str = $"  Statistics: {json}";
            WriteLine(str);
        }

        private void EventConsumer_OnPartitionsRevoked(object sender, List<TopicPartitionOffset> partitions)
        {
            string str = string.Empty;
            str = $"  Revoked partitions: [{string.Join(", ", partitions)}]";
            WriteLine(str);
            eventConsumer.Unassign();
        }

        private void EventConsumer_OnPartitionsAssigned(object sender, List<TopicPartitionOffset> partitions)
        {
            string str = string.Empty;
            str = $"  Assigned partitions : [{string.Join(", ", partitions)}]";
            WriteLine(str);
            eventConsumer.Assign(partitions);
        }

        private void EventConsumer_OnError(object sender, Handle.ErrorArgs e)
        {
            string str = string.Empty;
            str = $"  Error:{e.ErrorCode}  {e.Reason}";
            WriteLine(str);
        }

        private void EventConsumer_OnEndReached(object sender, TopicPartitionOffset e)
        {
            string str = string.Empty;
            str = $"  Reached end of topic: { e.Topic} partition: {e.Partition}, next message will be at offset {e.Offset} ";
            WriteLine(str);
        }

        private void EventConsumer_OnConsumerError(object sender, ErrorCode e)
        {
            string str = ($"  Consumer Error:{e}");
            WriteLine(str);
        }
        #endregion

        #region 读取一条kafka数据
        /// <summary>
        /// 读取一条kafka数据
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="msg"></param>
        private void EventConsumer_OnMessage(object sender, Message msg)
        {
            string text = Encoding.UTF8.GetString(msg.Payload, 0, msg.Payload.Length);

            string str = $"  Topic:={msg.Topic}    Partition:={msg.Partition}   Offset:={msg.Offset} ";
 
            WriteLine(str);
            WriteLine(text);

            if (text[0] == '$')
            {
                int index = text.IndexOf('$', 2);
                if (index > 0)
                { 
                    string companyId = text.Substring(1, index);
                    if(int.TryParse(companyId,out int cId))
                    {
                       if(  udpDataSendClientDict.ContainsKey(cId))
                        {
                            udpDataSendClientDict[cId].UdpSendMsg(text.Substring(index, text.Length - index));
                        }
                    }

                }
            }
            if (!enableAutoCommit && msg.Offset % 10 == 0)
            {
                str = ("  Commiting offset");
                WriteLine(str);
                eventConsumer.Commit(msg).GetAwaiter().GetResult();
                str = ("  Commited offset");
                WriteLine(str);
            }
        }
        #endregion


    }

}